package com.tron.sync.controller;

import com.tron.sync.entities.SyndaOrg;
import com.tron.sync.entities.SyndaOrgDoc;
import com.tron.sync.entities.SyndaUser;
import com.tron.sync.entities.SyndaUserDoc;
import com.tron.sync.service.ElkService;
import com.tron.sync.service.SyndaOrgService;
import com.tron.sync.service.SyndaUserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;

@Slf4j
public class KafkaMessageHandler extends Thread {
    private KafkaConsumer<String,String> kafkaConsumer;
    private String topic;

    private  SyndaOrgService syndaOrgService;

    private  SyndaUserService syndaUserService;

    private  ElkService elkService;

    public KafkaMessageHandler(KafkaConsumer<String,String> kafkaConsumer, String topic, SyndaOrgService syndaOrgService, SyndaUserService syndaUserService,  ElkService elkService){
        this.kafkaConsumer = kafkaConsumer;
        this.topic = topic;
        this.syndaOrgService = syndaOrgService;
        this.syndaUserService = syndaUserService;
        this.elkService = elkService;
    }

    @Override
    public void run() {
        log.info("============开始监听组织用户管理kafka消息.......topic为{}==========",topic);
        kafkaConsumer.subscribe(Collections.singletonList(topic));
        while(true){
            try {
                Thread.sleep(500);
                ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    log.info("Received message: (" + record.value());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void sync(Date date) {
        // 查询synda库中用户数据
        final List<SyndaUser> syndaUsers = syndaUserService.findByTime(date);

        // 全文检索中更新用户数据
        final ArrayList<SyndaUserDoc> syndaUserDocs = new ArrayList<>();
        syndaUsers.forEach(i -> {
            final SyndaUserDoc syndaUserDoc = new SyndaUserDoc();
            syndaUserDoc.setId(i.getId());
            syndaUserDoc.setName(i.getName());
            syndaUserDoc.setGender(i.getGender());
            syndaUserDoc.setStatus(i.getStatus());
            syndaUserDoc.setUserMobile(i.getMobile());
            syndaUserDoc.setTelephone(i.getTelephone());
            syndaUserDoc.setInsidePhone(i.getInsidePhone());
            syndaUserDoc.setJob(i.getJob());
            syndaUserDoc.setJobRank(i.getJobRank());
            syndaUserDoc.setUserNumber(i.getUserNumber());
            syndaUserDoc.setCategory(i.getCategory());
            syndaUserDoc.setUserType(i.getType());
            syndaUserDoc.setPinyin(i.getPinyin());
            syndaUserDoc.setOrgId(i.getGroupId());
            syndaUserDoc.setOrgType(i.getOrgType());
            syndaUserDoc.setPath(i.getPath());
            syndaUserDoc.setOrgName(i.getOrgName());
            try {
                syndaUserDoc.setHireDate(i.getHireDate().toInstant().toEpochMilli());
                syndaUserDoc.setJoinPartyDate(i.getJoinPartyDate().toInstant().toEpochMilli());
            } catch (Exception e) {

            }
            syndaUserDoc.setUserSort(i.getSort());
            syndaUserDoc.setUserIdentity(i.getUserIdentity());
            syndaUserDocs.add(syndaUserDoc);
        });
        elkService.saveUser(syndaUserDocs);

        // 查询synda库中机构数据
        final List<SyndaOrg> syndaOrgs = syndaOrgService.findByTime(date);

        // 全文检索中更新机构数据
        final ArrayList<SyndaOrgDoc> syndaOrgDocs = new ArrayList<>();
        syndaOrgs.forEach(i->{
            final SyndaOrgDoc syndaOrgDoc = new SyndaOrgDoc();
            syndaOrgDoc.setId(i.getId());
            syndaOrgDoc.setName(i.getName());
            syndaOrgDoc.setOrgAlias(i.getOrgAlias());
            syndaOrgDoc.setStatus(i.getStatus());
            syndaOrgDoc.setLevelId(i.getLevelId());
            syndaOrgDoc.setRoleIds(i.getRoleIds());
            syndaOrgDoc.setOrgType(i.getType());
            syndaOrgDoc.setAttribute(i.getAttribute());
            syndaOrgDoc.setOrgPid(i.getOrgPid());
            syndaOrgDoc.setOrgCode(i.getOrgCode());
            syndaOrgDoc.setRegion(i.getRegion());
            syndaOrgDoc.setBrevity(i.getBrevity());
            syndaOrgDoc.setInherit(i.getInherit());
            syndaOrgDoc.setSort(i.getSort());
            syndaOrgDoc.setClassIfy(i.getClassIfy());
            syndaOrgDoc.setPath(i.getPath());
            syndaOrgDocs.add(syndaOrgDoc);
        });
        elkService.saveOrg(syndaOrgDocs);
    }
}
